#ifndef __M_SERVER_H__
#define __M_SERVER_H__
#include <vector>
#include <algorithm>
#include <iostream>
#include <cassert>
#include <string>
#include <unordered_set>
#include <unistd.h>
#include <cstring>
#include <sys/types.h>          /* See NOTES */
#include <sys/socket.h>
#include <arpa/inet.h>
#include <thread>
#include <fcntl.h>
#include <sys/epoll.h>
#include <signal.h>
#include <functional>
#include <mutex>
#include <condition_variable>
#include <sys/timerfd.h>
#include <sys/eventfd.h>
/*---------simple logger------------*/

#define INF 0
#define DBG 1
#define ERR 2
#define LOG_LEVEL DBG

#define LOG(level, format, ...) do{\
        if (level < LOG_LEVEL) break;\
        time_t t = time(NULL);\
        struct tm *ltm = localtime(&t);\
        char tmp[32] = {0};\
        strftime(tmp, 31, "%H:%M:%S", ltm);\
        fprintf(stdout, "[%p %s %s:%d] " format "\n", (void*)pthread_self(), tmp, __FILE__, __LINE__, ##__VA_ARGS__);\
    }while(0)

#define INF_LOG(format, ...) LOG(INF, format, ##__VA_ARGS__)
#define DBG_LOG(format, ...) LOG(DBG, format, ##__VA_ARGS__)
#define ERR_LOG(format, ...) LOG(ERR, format, ##__VA_ARGS__)


/*---------缓冲区模块的实现-----------*/
const static uint64_t DEFAULT_BUFFER_SIZE = 1024;

class Buffer
{
public:
    Buffer():_buffer(DEFAULT_BUFFER_SIZE),_head(0),_tail(0){}
    
    //获取读位置
    char* readPos(){return begin() + _head;}
    //获取写位置
    char* writePos(){return begin() + _tail;}
    //可读大小
    uint64_t readableSize(){return _tail-_head;}
    //移动头部
    void moveRead(uint64_t len){
        assert(_head + len <= _tail);
        _head += len;
    }
    //移动尾部
    void moveWrite(uint64_t len){
        assert(_tail + len <= _buffer.size());
        _tail += len;
    }
    //写数据
    void write(const void* buffer,uint64_t len){
        modifyCapacity(len);
        const char*head=(const char*)buffer;
        const char*tail=(const char*)buffer+len;
        std::copy(head,tail,writePos());
        moveWrite(len);
    }
    void writeBuffer(Buffer& data){
        this->write(data.readPos(),data.readableSize());
    }

    //读数据
    void read(void* buffer,uint64_t len){
        assert(len<=readableSize());
        char*head=(char*)buffer;
        std::copy(readPos(),readPos()+len,head);   
        moveRead(len);
    }
    std::string getLine(){
        //get CRLF pos
        char* pos = (char*)memchr(readPos(),'\n',readableSize());
        std::string ret;    
        if(pos!=nullptr){
            uint64_t size=pos-readPos()+1;
            ret.resize(size);
            this->read(&ret[0],size);
        }
        return ret;
    }

    void clear(){_head=0;_tail=0;}
private:
    char* begin(){return &*_buffer.begin();}

    //确保有足够的空间进行写操作
    void modifyCapacity(uint64_t len){
        if(len + _tail < _buffer.size())
            return ;
        if( _buffer.size() - _tail + _head >= len){//calculate total size
            uint64_t size = readableSize();
            //copy forward
            std::copy(readPos(),writePos(),begin());
            _head=0;
            _tail=size;
        }
        else{//doubel capacity
            uint64_t sz=_buffer.size();
            _buffer.resize(sz+len);
        }
    }

private:
    std::vector<char>_buffer;
    uint64_t _head;
    uint64_t _tail;
};

/*-------Socket packaging-----------*/
static const size_t DEFAULT_LISTEN_NUM=1024;
class Socket
{
public:
    Socket(int sockfd):_sockfd(sockfd){};
    Socket(){create();}
    ~Socket(){shutDown();}
    bool create(){
        _sockfd=socket(AF_INET,SOCK_STREAM,IPPROTO_TCP);
        DBG_LOG("CREATE SOCKET");
        if(_sockfd < 0){
            ERR_LOG("SOCKED CREATE FAILED");
            return false;
        }
        return true;
    }

    int fd(){return _sockfd;}

    bool sockBind(const std::string& ip,uint16_t port){
        struct sockaddr_in addr;
        if(inet_pton(AF_INET,ip.c_str(),&(addr.sin_addr))!=1){
            ERR_LOG("IP CONVERT FAILED");
            return false;
        }
        addr.sin_port=htons(port);
        addr.sin_family=AF_INET;
        int ret = bind(_sockfd,(struct sockaddr*)&addr,sizeof(addr));
        if(ret < 0){
            ERR_LOG("BIND FAIL");
            return false;
        }
        DBG_LOG("BIND SUCCEED");
        return true;
    }

    bool sockConnect(const std::string& ip,uint16_t port){
        struct sockaddr_in addr;
        if(inet_pton(AF_INET,ip.c_str(),&(addr.sin_addr))!=1){
            ERR_LOG("IP CONVERT FAILED");
            return false;
        }
        addr.sin_port=htons(port);
        addr.sin_family=AF_INET;
        int ret = connect(_sockfd,(struct sockaddr*)&addr,sizeof(addr));
        if(ret < 0){
            ERR_LOG("CONNECT FAILED");
            return false;
        }
        DBG_LOG("CONNECT SUCCEED");
        return true;
    }
    bool sockListen(){
        if(listen(_sockfd,DEFAULT_LISTEN_NUM) != 0){
            ERR_LOG("LISTEN FAILED");
            return false;
        }
        DBG_LOG("LISTEN SUCCEED");
        return true;
    }

    int sockAccpet(){
        int new_sockfd=accept(_sockfd,nullptr,0);
        if(new_sockfd < 0){
            ERR_LOG("ACCEPT FAILED");
            return -1;
        }

        DBG_LOG("ACCEPT SUCCEED");
        return new_sockfd;
    }

    ssize_t sockSend(const void* buf,size_t len,int flag=0){
        ssize_t ret=send(_sockfd,buf,len,flag);
        if(ret<0){
            if(errno == EAGAIN || errno==EINTR){
                return 0;
            }
            ERR_LOG("SEND FAILED");
            return -1;
        }
        return ret;
    }

    ssize_t sockRecv(void* buf,size_t len,int flag=0){
        ssize_t ret = recv(_sockfd,buf,len,flag);
        if(ret<0){
            if(errno == EAGAIN || errno==EINTR){
                return 0;
            }
            ERR_LOG("RECIEVE FAILED");
            return -1;
        }
        return ret;
    }
    
    ssize_t noblockSend(void* buf,size_t len){
        return sockSend(buf,len,MSG_DONTWAIT);
    }

    ssize_t noblockRecv(void* buf,size_t len){
        return sockRecv(buf,len,MSG_DONTWAIT);
    }

    void shutDown(){
        if(_sockfd != -1){
            close(_sockfd);
            _sockfd=-1;
        }
    }

    void setSockNonblock(){
        int flag=fcntl(_sockfd,F_GETFL,0);
        fcntl(_sockfd,F_SETFL,flag|O_NONBLOCK);
    }

    void reuseAddress() {
        // int setsockopt(int fd, int leve, int optname, void *val, int vallen)
        int val = 1;
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, (void*)&val, sizeof(int));
        val = 1;
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEPORT, (void*)&val, sizeof(int));
    }

    bool createServer(uint16_t port
                    ,const std::string &ip = "0.0.0.0"
                    , bool block_flag = false){
        // if(create()==false)return false;
        reuseAddress();//reuse need before calling bind
        if(block_flag)setSockNonblock();
        if(sockBind(ip,port)==false)return false;
        if(sockListen()==false)return false;
        return true;
    }
    bool createClient(uint16_t port, const std::string &ip){
        if(create()==false)return false;
        if(sockConnect(ip,port)==false)return false;
        return true;
    }

private:
    int _sockfd=-1;
};

/*--------事件管理模块----------*/
class Poller;
class EventLoop;
class Channel
{
    using EventCallBack = std::function<void()>;
public:
    Channel(int fd,EventLoop* loop):_fd(fd),_loop(loop),_events(0),_revents(0){}
    const int Fd()const { return _fd; }
    const uint32_t events()const { return _events; }//获取想要监控的事件
    void setREvents(uint32_t events) { _revents = events; }//设置实际就绪的事件
    
    //实际调用回调的函数
    void handleEvent(){
        if(_revents & EPOLLERR)
            if(_error_callback)_error_callback();
        if(_revents & EPOLLHUP)
            if(_close_callback)_close_callback();
        
        if(_event_callback)
            _event_callback();

        if((_revents & EPOLLIN)||(_revents & EPOLLRDHUP))
            if(_read_callback)_read_callback();
        if(_revents & EPOLLOUT)
            if(_write_callback)_write_callback();
    }


    //处理事件相关的函数
    void remove();
    void update();

    bool readable(){return EPOLLIN|_events;}
    bool writeable(){return EPOLLOUT | _events;}
    
    void enableRead(){_events|=EPOLLIN;update();}
    void enableWrite(){_events|=EPOLLOUT;update();}
    void disableRead(){_events&=~EPOLLIN;update();}
    void disableWrite(){_events&=~EPOLLOUT;update();}
    void disableAll(){_events=0;update();}    
    
    //set callback
    void setWriteCallback(const EventCallBack& cb){_write_callback=cb;}
    void setReadCallback(const EventCallBack& cb){_read_callback=cb;}
    void setErrorCallback(const EventCallBack& cb){_error_callback=cb;}
    void setCloseCallback(const EventCallBack& cb){_close_callback=cb;}
    void setEventCallback(const EventCallBack& cb){_event_callback=cb;}

private:
    int _fd;
    EventLoop* _loop;
    uint32_t _events;    //正在监控的事件
    uint32_t _revents;   //监控事件中触发的事件

    EventCallBack _write_callback;
    EventCallBack _read_callback;
    EventCallBack _error_callback;
    EventCallBack _close_callback;
    EventCallBack _event_callback;
};

/*--------描述符事件监控模块--------*/
static const size_t EPOLL_EVENTS_NUM=1024;
class Poller
{
public:
    Poller(){
        _epfd  = epoll_create(EPOLL_EVENTS_NUM);
        if(_epfd<0){
            ERR_LOG("EPOLL CREATE FAILED");
            abort();
        }
    }
    ~Poller(){
        if (_epfd >= 0) { // 检查是否成功创建
            close(_epfd);
            _epfd = -1;
        }   
    }

    //开启监控函数，对标epoll_wait
    void pollerWait(std::vector<Channel*>&events){
        int n = epoll_wait(_epfd,_events,EPOLL_EVENTS_NUM,-1);
        if(n < 0){
            if(errno == EINTR)
                return;
            ERR_LOG("EPOLL WAIT ERROR:%s\n", strerror(errno));
            abort();
        }
        for(int i=0;i<n;i++){
            Channel* channel = (Channel*)_events[i].data.ptr;
            channel->setREvents(_events[i].events);
            events.push_back(channel);
        }
    }

    void updateEvent(Channel* channel){
        assert(channel);
        if(hasChannel(channel)==false){
            _channels.insert(channel);
            epollCtl(channel,EPOLL_CTL_ADD);
            DBG_LOG("ADD EPOLL");
        }
        else
            epollCtl(channel,EPOLL_CTL_MOD);
    }
    void removeEvent(Channel* channel){
        assert(channel);
        assert(hasChannel(channel) == true);
        _channels.erase(channel);
        epollCtl(channel,EPOLL_CTL_DEL);
    }

private:
    void epollCtl(Channel* channel,int opt){
        struct epoll_event ee;
        ee.events=channel->events();
        ee.data.ptr=(void*)channel;
        if(epoll_ctl(_epfd,opt,channel->Fd(),&ee) < 0){
            ERR_LOG("EPOLL CONTROL FAILED");
            return;
        }
    }

    bool hasChannel(Channel* channel){
        assert(channel);
        auto itr = _channels.find(channel);
        return itr != _channels.end();
    }
    
private:
    std::unordered_set<Channel*>_channels;
    int _epfd;
    struct epoll_event _events[EPOLL_EVENTS_NUM]; 
};
/*---------TimeWheel 模块-----------*/

using TaskFunc = std::function<void()>;
using ReleaseFunc = std::function<void()>;
class TimerTask{
    private:
        uint64_t _id;       // 定时器任务对象ID
        uint32_t _timeout;  //定时任务的超时时间
        bool _canceled;     // false-表示没有被取消， true-表示被取消
        TaskFunc _task_cb;  //定时器对象要执行的定时任务
        ReleaseFunc _release; //用于删除TimerWheel中保存的定时器对象信息
    public:
        TimerTask(uint64_t id, uint32_t delay, const TaskFunc &cb): 
            _id(id), _timeout(delay), _task_cb(cb), _canceled(false) {}
        ~TimerTask() { 
            if (_canceled == false) _task_cb(); 
            _release();     
        }
        void cancel() { _canceled = true; }
        void setRelease(const ReleaseFunc &cb) { _release = cb; }
        uint32_t delayTime() { return _timeout; }
};

class TimerWheel {
    private:
        using WeakTask = std::weak_ptr<TimerTask>;
        using PtrTask = std::shared_ptr<TimerTask>;
        int _tick;      //当前的秒针，走到哪里释放哪里，释放哪里，就相当于执行哪里的任务
        int _capacity;  //表盘最大数量---其实就是最大延迟时间
        std::vector<std::vector<PtrTask>> _wheel;
        std::unordered_map<uint64_t, WeakTask> _timers;

        EventLoop *_loop;
        int _timerfd;//定时器描述符--可读事件回调就是读取计数器，执行定时任务
        std::unique_ptr<Channel> _timer_channel;
    private:
        void removeTimer(uint64_t id) {
            auto it = _timers.find(id);
            if (it != _timers.end()) {
                _timers.erase(it);
            }
        }
        static int createTimerfd() {
            int timerfd = timerfd_create(CLOCK_MONOTONIC, 0);
            if (timerfd < 0) {
                ERR_LOG("TIMERFD CREATE FAILED!");
                abort();
            }
            //int timerfd_settime(int fd, int flags, struct itimerspec *new, struct itimerspec *old);
            struct itimerspec itime;
            itime.it_value.tv_sec = 1;
            itime.it_value.tv_nsec = 0;//第一次超时时间为1s后
            itime.it_interval.tv_sec = 1; 
            itime.it_interval.tv_nsec = 0; //第一次超时后，每次超时的间隔时
            timerfd_settime(timerfd, 0, &itime, NULL);
            return timerfd;
        }
        int readTimefd() {
            uint64_t times;
            //有可能因为其他描述符的事件处理花费事件比较长，然后在处理定时器描述符事件的时候，有可能就已经超时了很多次
            //read读取到的数据times就是从上一次read之后超时的次数
            int ret = read(_timerfd, &times, 8);
            if (ret < 0) {
                ERR_LOG("READ TIMEFD FAILED!");
                abort();
            }
            return times;
        }

        void onTime() {
            //根据实际超时的次数，执行对应的超时任务
            int times = readTimefd();
            for (int i = 0; i < times; i++) {
                _tick = (_tick + 1) % _capacity;
                _wheel[_tick].clear();
            }
        }
        void timerAddInLoop(uint64_t id, uint32_t delay, const TaskFunc &cb) {
            PtrTask pt(new TimerTask(id, delay, cb));
            pt->setRelease(std::bind(&TimerWheel::removeTimer, this, id));
            int pos = (_tick + delay) % _capacity;
            _wheel[pos].push_back(pt);
            _timers[id] = WeakTask(pt);
        }
        void timerRefreshInLoop(uint64_t id) {
            //通过保存的定时器对象的weak_ptr构造一个shared_ptr出来，添加到轮子中
            auto it = _timers.find(id);
            if (it == _timers.end()) {
                return;//没找着定时任务，没法刷新，没法延迟
            }
            PtrTask pt = it->second.lock();//lock获取weak_ptr管理的对象对应的shared_ptr
            int delay = pt->delayTime();
            int pos = (_tick + delay) % _capacity;
            _wheel[pos].push_back(pt);
        }
        void timerCancelInLoop(uint64_t id) {
            auto it = _timers.find(id);
            if (it == _timers.end()) {
                return;//没找着定时任务，没法刷新，没法延迟
            }
            PtrTask pt = it->second.lock();
            if (pt) pt->cancel();
        }
    public:
        TimerWheel(EventLoop *loop):_capacity(60), _tick(0), _wheel(_capacity), _loop(loop), 
            _timerfd(createTimerfd()), _timer_channel(new Channel(_timerfd,_loop)) {
            _timer_channel->setReadCallback(std::bind(&TimerWheel::onTime, this));
            _timer_channel->enableRead();//启动读事件监控
        }
        /*定时器中有个_timers成员，定时器信息的操作有可能在多线程中进行，因此需要考虑线程安全问题*/
        /*如果不想加锁，那就把对定期的所有操作，都放到一个线程中进行*/
        void timerAdd(uint64_t id, uint32_t delay, const TaskFunc &cb);
        //刷新/延迟定时任务
        void timerRefresh(uint64_t id);
        void timerCancel(uint64_t id);
        /*这个接口存在线程安全问题--这个接口实际上不能被外界使用者调用，只能在模块内，在对应的EventLoop线程内执行*/
        bool hasTimer(uint64_t id) {
            auto it = _timers.find(id);
            if (it == _timers.end()) {
                return false;
            }
            return true;
        }
};


//---------TODO这里的_timer是否有必要申明为unordered_map？？？？？？？？？？？？？？？？----------
// //时间轮的子模块
// class TimeTask
// {
//     using TimeCallBack = std::function<void()>;
// public:
//     TimeTask(uint64_t id
//             ,uint32_t timeout
//             ,const TimeCallBack&task
//             ,const TimeCallBack&release_wheel_task)
//             :_id(id)
//             ,_timeout(timeout)
//             ,_cancel(false)
//             ,_task(task)
//             ,_release_wheel_task(release_wheel_task){}
//     ~TimeTask(){
//         if(_cancel == false)
//             _task();
//         _release_wheel_task();
//     }
//     uint32_t delayTime(){return _timeout;}
// private:
//     uint64_t _id;
//     uint32_t _timeout;
//     bool _cancel;
//     TimeCallBack _task;
//     TimeCallBack _release_wheel_task;
// };
// class TimeWheel
// {
//     using TimeCallBack = std::function<void()>;
//     using TaskPtr = std::shared_ptr<TimeTask>;
//     using TaskWeakPtr = std::weak_ptr<TimeTask>;
// public:
//     TimeWheel(EventLoop* loop,int capacity = 60)
//     :_loop(loop)
//     ,_capacity(capacity)
//     ,_tick(0)
//     ,_wheel(capacity)
//     ,_timefd(createTimefd())
//     ,_channel(std::make_unique<Channel>(_loop,_timefd))
//     {
//         _channel->setReadCallback(std::bind(&TimeWheel::onTime,this));
//         _channel->enableRead();
//     }
// private:
//     void releaseTask(){
//     }
//     //创建配合epoll的timefd
//     int createTimefd(){
//         int timerfd = timerfd_create(CLOCK_MONOTONIC, 0);
//         if (timerfd < 0) {
//             ERR_LOG("TIMERFD CREATE FAILED!");
//             abort();
//         }
//         //int timerfd_settime(int fd, int flags, struct itimerspec *new, struct itimerspec *old);
//         struct itimerspec itime;
//         itime.it_value.tv_sec = 1;
//         itime.it_value.tv_nsec = 0;//第一次超时时间为1s后
//         itime.it_interval.tv_sec = 1; 
//         itime.it_interval.tv_nsec = 0; //第一次超时后，每次超时的间隔时
//         timerfd_settime(timerfd, 0, &itime, NULL);
//         return timerfd;
//     }
//     int onTime(){
//         uint64_t times;
//         if(read(_timefd,&time,8) < 0){
//             ERR_LOG("READ TIMEFD FAILED");
//             abort();
//         }
//         while(times--){
//             _tick = (_tick+1)%_capacity;
//             _wheel[_tick].clear();
//         }
//     }   
// private:
//     EventLoop* _loop;
//     int _capacity;
//     int _tick;
//     std::vector<std::vector<TaskPtr>>_wheel;
//     std::unordered_map<uint64_t, TaskWeakPtr> _timers;
//     int _timefd;
//     std::unique_ptr<Channel> _channel;
// };



/*---------EventLoop 模块-----------*/
class EventLoop
{
    using TaskFunc = std::function<void()>;
public:
    EventLoop()
    :_eventfd(CreateEventFd())
    ,_thread_id(std::this_thread::get_id())
    ,_channel(std::make_unique<Channel>(_eventfd,this))
    ,_wheel(this){
        _channel->setReadCallback(std::bind(&EventLoop::readEventfd,this));
        _channel->enableRead();
    }

    void start(){
        while(1){
            std::vector<Channel*>actives;
            _poller.pollerWait(actives);

            for(auto& cl:actives){
                cl->handleEvent();
            }

            std::vector<TaskFunc>new_tasks;
            {
                std::unique_lock<std::mutex>lock(_mtx);
                _tasks.swap(new_tasks);
            }
            for(auto& func:new_tasks){
                func();
            }
        }

    }

    void runInLoop(const TaskFunc&cb){
        if(isInLoop()){
            cb();
        }
        else
            queueInLoop(cb);
    }

    void queueInLoop(const TaskFunc&cb){
        {
            std::unique_lock<std::mutex>lock(_mtx);
            _tasks.push_back(cb);
        }
        writeEventfd();
    }
  
    void assertInLoop(){
        assert(std::this_thread::get_id() == _thread_id);
    }

    void updateEvent(Channel* channel){return _poller.updateEvent(channel);}
    void removeEvent(Channel* channel){return _poller.removeEvent(channel);}
    void timerAdd(uint64_t id,uint32_t delay,const TaskFunc& cb){_wheel.timerAdd(id,delay,cb);}
    void timerRefresh(uint64_t id){_wheel.timerRefresh(id);}
    void timerCancel(uint64_t id){_wheel.timerCancel(id);}
    bool hasTimer(uint64_t id) { return _wheel.hasTimer(id);}

private:
    bool isInLoop(){
        return std::this_thread::get_id() == _thread_id;
    }


    static int CreateEventFd() {
        int efd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
        if (efd < 0) {
            ERR_LOG("CREATE EVENTFD FAILED!!");
            abort();//让程序异常退出
        }
        return efd;
    }

    void readEventfd(){
        uint64_t buf = 0;
        int ret = read(_eventfd,&buf,sizeof(buf));
        if(ret < 0){
            if(errno == EINTR || errno == EAGAIN){
                return;
            }
            ERR_LOG("EVENTFD READ FAILED");
            abort();    
        }
    }

    void writeEventfd(){
        uint64_t val = 1;
        int ret = write(_eventfd,&val,sizeof(val));
        if(ret < 0){
            if(errno == EINTR || errno == EAGAIN){
                return;
            }
            ERR_LOG("EVENTFD WRITE FAILED");
            abort();
        }
    }
private:
    int _eventfd;
    std::thread::id _thread_id;
    std::unique_ptr<Channel>_channel;
    std::mutex _mtx;
    std::vector<TaskFunc>_tasks;
    Poller _poller;
    TimerWheel _wheel;

};

/*与EventLoop 配套的thread模块*/
//one thread one loop
class LoopThread
{
public:
    LoopThread()
    :_loop(nullptr)
    ,_thread(&LoopThread::entryThread,this)
    {}

    EventLoop* getLoop(){
        std::unique_lock<std::mutex>lock(_mtx);
        _cond.wait(lock,[&]{return _loop != nullptr;});
        return _loop;
    }

private:
    void entryThread(){
        EventLoop loop;
        {
            std::unique_lock<std::mutex>lock(_mtx);
            _loop = &loop;
            _cond.notify_all();
        }
        _loop->start();
    }
private:
    std::mutex _mtx;
    std::condition_variable _cond;
    EventLoop* _loop;
    std::thread _thread;
};


class LoopThreadPool
{
public:
    LoopThreadPool(EventLoop* baseloop):_thread_count(0),_next_index(0),_baseloop(baseloop){}
    void create(){
        _loops.resize(_thread_count);
        _threads.resize(_thread_count);
        for(size_t i = 0;i<_thread_count;i++){
            _threads[i] = std::make_unique<LoopThread>();
            _loops[i] = _threads[i]->getLoop();
        }
    }
    void setThreadCount(int count){
        _thread_count = count;
    }

    EventLoop* nextLoop(){
        if(_thread_count == 0)
            return _baseloop;
        EventLoop* loop = _loops[_next_index];
        _next_index = (1 + _next_index)%_thread_count;
        return loop;
    }

private:
    size_t _thread_count;
    int _next_index;     
    EventLoop* _baseloop;
    std::vector<std::unique_ptr<LoopThread>>_threads;
    std::vector<EventLoop*>_loops;
};


class Any{
private:
    class holder {
        public:
            virtual ~holder() {}
            virtual const std::type_info& type() = 0;
            virtual holder *clone() = 0;
    };
    template<class T>
    class placeholder: public holder {
        public:
            placeholder(const T &val): _val(val) {}
            // 获取子类对象保存的数据类型
            virtual const std::type_info& type() { return typeid(T); }
            // 针对当前的对象自身，克隆出一个新的子类对象
            virtual holder *clone() { return new placeholder(_val); }
        public:
            T _val;
    };
    holder *_content;
public:
    Any():_content(NULL) {}
    template<class T>
    Any(const T &val):_content(new placeholder<T>(val)) {}
    Any(const Any &other):_content(other._content ? other._content->clone() : NULL) {}
    ~Any() { delete _content; }

    Any &swap(Any &other) {
        std::swap(_content, other._content);
        return *this;
    }

    // 返回子类对象保存的数据的指针
    template<class T>
    T *get() {
        //想要获取的数据类型，必须和保存的数据类型一致
        assert(typeid(T) == _content->type());
        return &((placeholder<T>*)_content)->_val;
    }
    //赋值运算符的重载函数
    template<class T>
    Any& operator=(const T &val) {
        //为val构造一个临时的通用容器，然后与当前容器自身进行指针交换，临时对象释放的时候，原先保存的数据也就被释放
        Any(val).swap(*this);
        return *this;
    }
    Any& operator=(const Any &other) {
        Any(other).swap(*this);
        return *this;
    }

};
/*--------Connection模块的实现-------*/

typedef enum{DISCONNECTED,CONNECTED,DISCONNECTING,CONNECTING}ConnStatus;

class Connection :public std::enable_shared_from_this<Connection>
{
    using ConnectionPtr = std::shared_ptr<Connection>;

    using ConnectedCallback = std::function<void(const ConnectionPtr&)>;
    using MessageCallback = std::function<void(const ConnectionPtr&,Buffer*)>;
    using ClosedCallback = std::function<void(const ConnectionPtr&)>;
    using AnyEventCallback = std::function<void(const ConnectionPtr&)>;
public:
    Connection(uint64_t conn_id,int sockfd,EventLoop* loop)
    :_conn_id(conn_id)
    ,_sockfd(sockfd)
    ,_enable_inactive_release(false)
    ,_loop(loop)
    ,_channel(sockfd,loop)
    ,_status(CONNECTING)
    ,_socket(sockfd)
    {
        _channel.setCloseCallback(std::bind(&Connection::handleClose,this));
        _channel.setReadCallback(std::bind(&Connection::handleRead,this));
        _channel.setErrorCallback(std::bind(&Connection::handleError,this));
        _channel.setEventCallback(std::bind(&Connection::handleAnyEvent,this));
        _channel.setWriteCallback(std::bind(&Connection::handleWrite,this));
    }
    ~Connection(){ DBG_LOG("RELEASE CONNECTION:%p", this);};
    int fd(){return _sockfd;}
    int id(){return _conn_id;}
    void setContext(const Any& context){_context = context;}
    const Any&context(){return _context;}
    void setConnectedCallback(const ConnectedCallback&cb){_connected_callback=cb;}
    void setMessageCallback(const MessageCallback&cb){_message_callback=cb;}
    void setClosedCallback(const ClosedCallback&cb){_close_callback=cb;}
    void setAnyEventCallback(const AnyEventCallback&cb){_any_event_callback=cb;}
    void setServerCloseCallback(const ClosedCallback&cb){_server_close_callback = cb;}
    void established(){
        _loop->runInLoop(std::bind(&Connection::establishedInLoop,this));
    }
    void connectionSend(const char *data, size_t len){
        Buffer buf;
        buf.write(data,len);
        _loop->runInLoop(std::bind(&Connection::connectionSendInLoop,this,buf));
    }
    void shutdown(){_loop->runInLoop(std::bind(&Connection::shutdownInLoop,this));}
    void release(){_loop->runInLoop(std::bind(&Connection::releaseInLoop,this));}
    void enableInactiveRelease(int sec){_loop->runInLoop(std::bind(&Connection::enableInactiveReleaseInLoop,this,sec));}
    void cancelInactiveRelease(){_loop->runInLoop(std::bind(&Connection::cancelInactiveReleaseInLoop,this));}
    void upgrade(const Any& context
                ,const ConnectedCallback& connCb
                ,const MessageCallback& messCb
                ,const ClosedCallback& closeCb
                ,const AnyEventCallback& anyEventCb
                )
    {
        _loop->assertInLoop();
        _loop->runInLoop(std::bind(&Connection::upgradeInLoop,this,context,connCb,messCb,closeCb,anyEventCb));
    }


private:
    void handleRead(){
        char buf[65536];
        ssize_t ret = _socket.noblockRecv(buf,65535);
        if(ret < 0){
            return shutdownInLoop();
        }
        _in_buffer.write(buf,ret);
        if(_in_buffer.readableSize()>0){
            if(_message_callback)_message_callback(shared_from_this(),&_in_buffer);
        }

    }
    void handleWrite(){
        ssize_t ret = _socket.noblockSend(_out_buffer.readPos(),_out_buffer.readableSize());
        if(ret < 0){
            if(_in_buffer.readableSize()>0){
                if(_message_callback)_message_callback(shared_from_this(),&_in_buffer);
            }
            return release();
        }
        _out_buffer.moveRead(ret);
        if(_out_buffer.readableSize() == 0){
            _channel.disableWrite();
            if(_status == DISCONNECTING)
                return release();
        }
        
    }
    void handleClose(){
        if(_in_buffer.readableSize()>0){
            if(_message_callback)_message_callback(shared_from_this(),&_in_buffer);
        }
        release();
    }
    void handleError(){
        handleClose();
    }
    void handleAnyEvent(){
        if (_enable_inactive_release == true){  _loop->timerRefresh(_conn_id); }
        if (_any_event_callback){  _any_event_callback(shared_from_this()); }
    }

    void establishedInLoop(){
        assert(_status == CONNECTING);
        _status = CONNECTED;
        _channel.enableRead();
        if(_connected_callback)_connected_callback(shared_from_this());
    }
    void releaseInLoop(){//真正释放连接的接口
        _channel.remove();
        _socket.shutDown();
        if(_close_callback)_close_callback(shared_from_this());
        if (_loop->hasTimer(_conn_id)) cancelInactiveReleaseInLoop();
        if(_server_close_callback)_server_close_callback(shared_from_this());

    }
    void connectionSendInLoop(Buffer& buffer){
        if(_status == DISCONNECTED)return ;
        _out_buffer.writeBuffer(buffer);
        if(_channel.writeable() == false){
            _channel.enableWrite();
        }
    }
    void shutdownInLoop(){
        _status = DISCONNECTING;
        if(_out_buffer.readableSize()>0&&_channel.writeable() == false){
            _channel.enableWrite();
        }
        if(_out_buffer.readableSize() == 0)
            release();
    }
    void enableInactiveReleaseInLoop(int sec){
        _enable_inactive_release=true;
        if(_loop->hasTimer(_conn_id))
            _loop->timerRefresh(_conn_id);
        else    
            _loop->timerAdd(_conn_id,sec,std::bind(&Connection::release,this));

    }
    void cancelInactiveReleaseInLoop(){
        _enable_inactive_release = false;
        if(_loop->hasTimer(_conn_id))
            _loop->timerCancel(_conn_id);
    }
    void upgradeInLoop(const Any& context
                ,const ConnectedCallback& connCb
                ,const MessageCallback& messCb
                ,const ClosedCallback& closeCb
                ,const AnyEventCallback& anyEventCb)
    {
        _context = context;
        _connected_callback = connCb;
        _message_callback = messCb;
        _close_callback = closeCb;
        _any_event_callback = anyEventCb;
    }

private:
    uint64_t _conn_id;
    int _sockfd;
    bool _enable_inactive_release;
    EventLoop* _loop;
    ConnStatus _status;
    Channel _channel;
    Socket _socket;
    Buffer _in_buffer;
    Buffer _out_buffer;
    Any _context;

    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    ClosedCallback _close_callback;
    ClosedCallback _server_close_callback;
    AnyEventCallback _any_event_callback;

};

/*------对监听套接字进行管理Acceptor----*/
class Acceptor
{
    using CallbackFunc = std::function<void(int)>;
public:
    Acceptor(EventLoop* loop,uint16_t port)
    :_socket(),_loop(loop),_channel(_socket.fd(),loop)
    {
        bool ret = _socket.createServer(port);
        assert(ret == true);
        _channel.setReadCallback(std::bind(&Acceptor::handleRead,this));
    }
    
    void setAcceptorCallback(const CallbackFunc&cb){_acceptor_callback = cb;}
    void Listen(){_channel.enableRead();}
private:

    void handleRead(){
        int newfd = _socket.sockAccpet();
        if(newfd < 0){
            ERR_LOG("%s",strerror(errno));
            return;
        }
        if(_acceptor_callback)_acceptor_callback(newfd);
    }
private:
    Socket _socket;
    EventLoop* _loop;
    Channel _channel;
    CallbackFunc _acceptor_callback;
};

/*-------总的整合模块TcpServer---------*/
class TcpServer
{
    using ConnectionPtr = std::shared_ptr<Connection>;

    using Functor = std::function<void()>;
    using MessageCallback = std::function<void(ConnectionPtr,Buffer*)>;
    using ClosedCallback = std::function<void(ConnectionPtr)>;
    using ConnectedCallback = std::function<void(ConnectionPtr)>;
    using AnyEventCallback = std::function<void(ConnectionPtr)>;
public:
    TcpServer(int port)
    :_next_id(0)
    ,_port(port)
    ,_timeout(0)
    ,_enable_inactive_release(false)
    ,_baseloop()
    ,_acceptor(&_baseloop,port)
    ,_pool(&_baseloop)
    {
        _acceptor.setAcceptorCallback(std::bind(&TcpServer::newConnection,this,std::placeholders::_1));
        _acceptor.Listen();
    }

    void setConnectedCallback(const ConnectedCallback&cb){_connected_callback=cb;}
    void setMessageCallback(const MessageCallback&cb){_message_callback=cb;}
    void setClosedCallback(const ClosedCallback&cb){_close_callback=cb;}
    void setAnyEventCallback(const AnyEventCallback&cb){_any_event_callback=cb;}

    void enableInactiveRelease(int timeout){
        if(_enable_inactive_release != true){
            _enable_inactive_release = true;
            _timeout = timeout;
            for(auto& [a,b]:_conns){
                b->enableInactiveRelease(timeout);
            }
        }
    }

    void start(){
        _pool.create();
        _baseloop.start();
    }

    void addTimeTask(int timeout,const Functor& func){
        _baseloop.runInLoop(std::bind(&TcpServer::addTimeTaskInloop,this,timeout,func));
    }
    
    void setThreadCount(int count){
        _pool.setThreadCount(count);
    }

private:
    
    void addTimeTaskInloop(int timeout,const Functor& func){
        ++_next_id;
        _baseloop.timerAdd(_next_id,timeout,func);
    }


    void removeConnectionInloop(const ConnectionPtr& conn_ptr){
        auto itr = _conns.find(conn_ptr->id());
        if(itr == _conns.end()){
            return;
        }
        _conns.erase(itr);
    }

    void removeConnetcion(const ConnectionPtr& conn_ptr){
        _baseloop.runInLoop(std::bind(&TcpServer::removeConnectionInloop,this,conn_ptr));
    }

    void newConnection(int fd){
        printf("NewConnection\n");
        _next_id++;
        ConnectionPtr conn = std::make_shared<Connection>(_next_id,fd,_pool.nextLoop());
        
        conn->setAnyEventCallback(_any_event_callback);
        conn->setClosedCallback(_close_callback);
        conn->setConnectedCallback(_connected_callback);
        conn->setMessageCallback(_message_callback);
        conn->setServerCloseCallback(std::bind(&TcpServer::removeConnetcion,this,std::placeholders::_1));

        if(_enable_inactive_release)conn->enableInactiveRelease(_timeout);
        conn->established();
        _conns[_next_id] = conn;
    }

private:
    uint64_t _next_id;
    int _port;
    int _timeout;
    bool _enable_inactive_release;
    EventLoop _baseloop;
    Acceptor _acceptor;
    LoopThreadPool _pool;
    std::unordered_map<uint64_t ,ConnectionPtr>_conns;


    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    ClosedCallback _close_callback;
    AnyEventCallback _any_event_callback;

};


/*------Channel 模块的函数----------*/
//更新事件
void Channel::update(){_loop->updateEvent(this);}
//移除事件
void Channel::remove(){_loop->removeEvent(this);}
/*---------------------------------*/



/*------TimeWheel 模块的函数--------*/
void TimerWheel::timerAdd(uint64_t id, uint32_t delay, const TaskFunc &cb) {
    _loop->runInLoop(std::bind(&TimerWheel::timerAddInLoop, this, id, delay, cb));
}
//刷新/延迟定时任务
void TimerWheel::timerRefresh(uint64_t id) {
    _loop->runInLoop(std::bind(&TimerWheel::timerRefreshInLoop, this, id));
}
void TimerWheel::timerCancel(uint64_t id) {
    _loop->runInLoop(std::bind(&TimerWheel::timerCancelInLoop, this, id));
}

/*---------------------------------*/


class NetWork {
    public:
        NetWork() {
            DBG_LOG("SIGPIPE INIT");
            signal(SIGPIPE, SIG_IGN);
        }
};
static NetWork nw;
#endif